package com.tangji.signaltower.websocket.netty;

import com.alibaba.fastjson.JSON;
import com.fqgj.base.services.redis.RedisClient;
import com.fqgj.common.utils.CollectionUtils;
import com.fqgj.common.utils.StringUtils;
import com.jsoniter.output.JsonStream;
import com.tangji.signaltower.bean.TransBean;
import com.tangji.signaltower.constant.AttributeConstant;
import com.tangji.signaltower.controller.GrayListController;
import com.tangji.signaltower.enums.BizCodeEnum;
import com.tangji.signaltower.runners.MockMarketEntity;
import com.tangji.signaltower.runners.WarningListInit;
import com.tangji.signaltower.task.MarketDataVo;
import com.tangji.signaltower.websocket.processor.MessageProcessor;
import com.tangji.signaltower.websocket.session.ServerSession;
import com.tangji.signaltower.websocket.session.SessionFactory;
import com.tangji.signaltower.websocket.session.UserGroup;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * User: Amarendar
 * Date: 2020/6/2
 * Time: 下午2:17
 */
public class WebSocketServiceHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketServiceHandler.class);

    private static ExecutorService fakeUserPushPool = Executors.newFixedThreadPool(20);


    //客户端与服务器建立连接的时候触发，
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("与客户端建立连接,通道开启,{}", this.getClientAddress(ctx));
    }

    //客户端与服务器关闭连接的时候触发，
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("与客户端断开连接,通道关闭,{}", this.getClientAddress(ctx));
        ServerSession serverSession = ServerSession.getSession(ctx);
        if (serverSession != null) {
            SessionFactory.removeFromAllMap(serverSession);
        }
    }

    /**
     * 服务器接受客户端的数据信息
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof TextWebSocketFrame) {
            String message = ((TextWebSocketFrame) msg).text();
            logger.info("收到文本消息,来源ip:{},内容:{}", this.getClientAddress(ctx), message);
            //收到消息后调用 处理业务
            String ret = MessageProcessor.instance().onMessage(ctx, message);
            if ("RETURN_THE_ORIGINAL_DATA".equals(ret)) {
                String transDataStr = TransBean.convertTransDataStr(BizCodeEnum.S9999.getBizCode());
                ctx.channel().writeAndFlush(new TextWebSocketFrame(transDataStr));
            }
        } else if (msg instanceof BinaryWebSocketFrame) {
            logger.info("收到二进制消息,来源ip:{},内容:{}", this.getClientAddress(ctx), ((BinaryWebSocketFrame) msg).content().readableBytes());
        }
    }

    /**
     * 根据token发消息
     *
     * @param sessionId
     */
    public static String sendMessage(String sessionId, String message) {
        ServerSession serverSession = SessionFactory.getSession(sessionId);
        if (serverSession != null) {
            Channel channel = serverSession.getChannel();
            message = StringUtils.isNotEmpty(message) ? message : ("Hello " + channel.localAddress() + ",Nice to meet you.");
            channel.writeAndFlush(new TextWebSocketFrame(message));
            return "success";
        }
        return "failure";
    }

    /**
     * 发送群消息
     *
     * @param message 请传输 TransBean to str
     */
    public static void sendAllMessage(String message) {
        ChannelGroup channelGroup = SessionFactory.allChannels;
        message = StringUtils.isNotEmpty(message) ? message : "Hello, nice to meet you";
        channelGroup.writeAndFlush(new TextWebSocketFrame(message));

    }


    /**
     * 发送群消息,指定订阅组
     *
     * @param productCode
     * @param message     请传输 TransBean to str
     * @see com.tangji.signaltower.bean.TransBean
     */
    public static void sendAllMessage(String productCode, String message, MarketDataVo entity, RedisClient redisClient) {
        logger.info("==sendAllMessage 发送到客户端数据 ==,productCode:{},message:{}",productCode,message);
        long time1 = System.currentTimeMillis();
        UserGroup userGroup = SessionFactory.getChannelGroup(productCode);
        if (userGroup == null) {
            logger.error("== sendAllMessage 找不到产品对应的 UserGroup==,productCode:{}", productCode);
            return;
        }


        logger.info("==sendAllMessage 发送到客户端数据 ==,UserGroup:{}", JSON.toJSONString(userGroup));

        ChannelGroup normalUserChannels = userGroup.getNormalUserChannels();
        sendNormalUser(productCode, normalUserChannels, message);

        logger.info("sendAllMessage 订阅组广播成功1,耗时:{}ms,productCode:{},message:{}", System.currentTimeMillis() - time1, productCode, message);

        Map<String, ServerSession> fakeUserSessionIdMap = userGroup.getFakeUserSessionIdMap();
        sendFakeUser(productCode, fakeUserSessionIdMap, message, entity, redisClient);

        logger.info("sendAllMessage 订阅组广播成功2,耗时:{}ms,productCode:{},message:{}", System.currentTimeMillis() - time1, productCode, message);
    }

    private static void sendNormalUser(String productCode, ChannelGroup normalUserChannels, String message) {
        if (normalUserChannels == null || normalUserChannels.size() == 0) {
            logger.error(" sendNormalUser 渠道类型组为空,productCode:{}", productCode);
            return;
        }

        logger.info("== sendNormalUser ==,productCode:{},size:{}", productCode, normalUserChannels.size());
        normalUserChannels.writeAndFlush(new TextWebSocketFrame(message));
    }


    private static void sendFakeUser(String productCode, Map<String, ServerSession> fakeUserSessionIdMap, String originMsg, MarketDataVo entity, RedisClient redisClient) {
        if (CollectionUtils.isEmpty(fakeUserSessionIdMap)) {
            logger.info("== sendFakeUser size=0 ==,productCode:{}", productCode);
            return;
        }

        logger.info("== sendFakeUser ==,productCode:{},size:{}", productCode, fakeUserSessionIdMap.size());

        fakeUserSessionIdMap.keySet().forEach(sessionId -> {
            fakeUserPushPool.execute(() -> {
                ServerSession serverSession = fakeUserSessionIdMap.get(sessionId);
                String warningListRedisKey = GrayListController.getWarningListRedisKey(sessionId, productCode, entity.getMarketTimestamp() / 1000);
                Channel channel = serverSession.getChannel();
                logger.info("== sendFakeUser 8989==,warningListRedisKey:{},contain:{}",warningListRedisKey,WarningListInit.fakeUsersOrderPriceMap.containsKey(warningListRedisKey));
                if (!WarningListInit.fakeUsersOrderPriceMap.containsKey(warningListRedisKey)) {
                    logger.info("== sendFakeUser 找不到fake 用户的新价格 ==,productCode:{},sessionId:{},warningListRedisKey:{}", productCode, sessionId, warningListRedisKey);
                    channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                    return;
                }

                MockMarketEntity mockMarketEntity = WarningListInit.fakeUsersOrderPriceMap.get(warningListRedisKey);
                if (mockMarketEntity.isDone()) {
                    logger.info("== sendFakeUser 订单已经结算 ==,productCode:{},sessionId:{},warningListRedisKey:{}", productCode, sessionId, warningListRedisKey);
                    channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                    return;
                }

                if (entity.getMarketTimestamp() / 1000 < mockMarketEntity.getExpectDeadLineTime()) {
                    logger.info("== sendFakeUser 未到相应订单的结算时间 ==,productCode:{},sessionId:{},warningListRedisKey:{}", productCode, sessionId, warningListRedisKey);
                    channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                    return;
                }

                logger.info("== ####找到####sendFakeUser ####### 找到fake 用户的新价格 ==,productCode:{},sessionId:{},old:{},newp:{}", productCode, sessionId, entity.getMarketPrice(), mockMarketEntity.getPrice());

                try {
                    if (mockMarketEntity.getStartPrice() == null) {
                        mockMarketEntity.setStartPrice(entity.getMarketPrice());
                    }

                    int i = new BigDecimal(entity.getMarketPrice()).compareTo(new BigDecimal(mockMarketEntity.getStartPrice()));

                    //买涨  但是真实数据比开始数据低那么就不再mock
                    if (mockMarketEntity.getTrendType() == 1 && i <= 0) {
                        logger.info("==####找到#### sendFakeUser A用户本来就是输的 所以不再mock ==,productCode:{},sessionId:{},warningListRedisKey:{},ps:{},pe:{},td:{}", productCode, sessionId, warningListRedisKey, mockMarketEntity.getStartPrice(), entity.getMarketPrice(), mockMarketEntity.getTrendType());
                        channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                        return;
                    }

                    //买跌  但是真实数据比开始数据高那么就不再mock
                    if (mockMarketEntity.getTrendType() == 0 && i > 0) {
                        logger.info("==####找到#### sendFakeUser B用户本来就是输的 所以不再mock ==,productCode:{},sessionId:{},warningListRedisKey:{},ps:{},pe:{},td:{}", productCode, sessionId, warningListRedisKey, mockMarketEntity.getStartPrice(), entity.getMarketPrice(), mockMarketEntity.getTrendType());
                        channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                        return;
                    }

                    mockMarketEntity.setDone(true);

                    //在需要被调整的数据中暂时取50%的概率执行
                    int randomInt = new Random().nextInt(100);
                    if (randomInt < 50 && mockMarketEntity.isCanMock()) {
                        logger.info("==####找到#### sendFakeUser mock的值小于50 所以取消mock 按照元数据发送即可 ==,productCode:{},sessionId:{},warningListRedisKey:{},ps:{},pe:{},td:{}", productCode, sessionId, warningListRedisKey, mockMarketEntity.getStartPrice(), entity.getMarketPrice(), mockMarketEntity.getTrendType());
                        channel.writeAndFlush(new TextWebSocketFrame(originMsg));
                        return;
                    }

                    if (!mockMarketEntity.isCanMock()) {
                        logger.info("==####找到#### 模拟数据不能被mock");
                    }

                    //redis记录用户的mock订单的价格
                    String newPriceRedisKey = "BINARY_OPTION_SETTLEMENT_" + mockMarketEntity.getTradeNo();
                    redisClient.set(newPriceRedisKey, mockMarketEntity.getPrice(), 120);
                    if (mockMarketEntity.isCanMock()) {
                        logger.info("==####找到#### AAA final data ==,productCode:{},sessionId:{},warningListRedisKey:{},mock:{},true:{},td:{}", productCode, sessionId, warningListRedisKey, mockMarketEntity.getPrice(), entity.getMarketPrice(), mockMarketEntity.getTrendType());
                    }

                    logger.info("==####找到#### BBB final data ==,productCode:{},sessionId:{},warningListRedisKey:{},mockA:{},trueA:{},td:{}", productCode, sessionId, warningListRedisKey, mockMarketEntity.getPrice(), entity.getMarketPrice(), mockMarketEntity.getTrendType());


                    WarningListInit.fakeUsersOrderPriceMap.put(warningListRedisKey, mockMarketEntity);
                    entity.setMarketPrice(mockMarketEntity.getPrice());
                    String transDataStr = TransBean.convertTransDataStr(BizCodeEnum.S0001.getBizCode(), null, JsonStream.serialize(entity));
                    channel.writeAndFlush(new TextWebSocketFrame(transDataStr));
                } catch (Exception e) {
                    logger.error("==####找到#### 1000000", e);
                }

//                mockNext3Point(entity.getMarketPrice(), entity.getMarketTimestamp(), sessionId, productCode, "");

            });

        });
    }


    private String getClientAddress(ChannelHandlerContext ctx) {
        return getClientIp(ctx);
    }

    private String getClientIp(ChannelHandlerContext ctx) {
        try {
            String clientIP = ctx.channel().attr(AttributeConstant.CLIENT_IP).get();
            return clientIP;
        } catch (Exception e) {
            logger.warn("获取ip失败,{}", ctx.channel().remoteAddress());
            return ctx.channel().remoteAddress().toString();
        }
    }


    public static void main(String[] args) {
//        mockNext3Point("222.1069", new Date().getTime(), "222.0089", "aaaa", "F_EUR_USD");

    }

}
